package com.corn.flink.lesson3;

import org.apache.flink.api.common.typeinfo.IntegerTypeInfo;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2023/2/23 18:02
 */

public class FlinkMapAndFlatMapFilterDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Integer> ds = env.fromCollection(
                IntStream.range(0, 10).boxed().collect(Collectors.toList())
        );
        // 元素 * 2
//        ds.map(i -> i * 2)
//                .print();
        // filter
//        ds.filter(i -> i > 5)
//                .print();

        ds.flatMap((Integer i, Collector<Integer> collector) -> {
            if (i == 3) {
                collector.collect(i * 50);
            }
        }).returns(Types.INT)
                .print();
        env.execute();
    }
}
